package stc.skymobi.netty.handler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import org.apache.commons.collections.Closure;
import org.jocean.util.ExceptionUtils;
import org.jocean.util.MBeanRegisterSupport;
import org.jocean.util.Visitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import stc.skymobi.ebus.EventBus;
import stc.skymobi.netty.transport.TransportUtils;
import stc.skymobi.netty.transport.Validateable;
import stc.skymobi.netty.transport.endpoint.AbstractEndpoint;
import stc.skymobi.netty.transport.endpoint.Endpoint;
import stc.skymobi.netty.transport.endpoint.EndpointFactory;
import stc.skymobi.util.Identifyable;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;

@ChannelHandler.Sharable
public class TcpRequestHandler extends ChannelInboundHandlerAdapter {

    public static interface ChannelMXBean {

        public String getId();

        public boolean isConnected();

        public String getRemoteIp();

        public int getRemotePort();

        public String getLocalIp();

        public int getLocalPort();
        
        public String invokeRemoteMBeanOperation(
        		final String objectName, 
        		final String operationName, 
        		final String[] params);
    }

    private static Logger logger = LoggerFactory.getLogger(TcpRequestHandler.class);

    private String eventRequestReceived;
    private String eventSessionOpened;
    private String eventSessionClosed;

    private EventBus eventBus;
    private Map<Object, Closure> uuidMap;

    private EndpointFactory endpointFactory = null;

    private final MBeanRegisterSupport _mbeanSupport;

    public TcpRequestHandler() {
        this("public:class=transport.tcp.netty.server,name=inbound");
    }

    public TcpRequestHandler(final String objectNamePrefix) {
        this._mbeanSupport = new MBeanRegisterSupport(objectNamePrefix, null);
    }

    public void destroy() {
        this._mbeanSupport.destroy();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        if (!(cause instanceof IOException)) {
            logger.error("exceptionCaught: ", cause);
        } else {
            logger.debug("exceptionCaught: ", cause);
        }
        ctx.channel().close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        logger.debug("messageReceived {}", msg);
        if (msg instanceof Validateable) {
            Object result = ((Validateable) msg).validate();
            if (result != null) {
                ctx.channel().writeAndFlush(result);
                logger.info("request {} validate failed,return resp {}", msg, result);
                return;
            }
        }
        Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
        if (null != endpoint) {
            endpoint.messageReceived(ctx, msg);
        }
        if (eventBus != null)
            eventBus.fireEvent(eventRequestReceived, msg);
        if (uuidMap != null) {
            if (msg instanceof Identifyable) {
                UUID uuid = ((Identifyable) msg).getIdentification();
                if (uuidMap.containsKey(uuid)) {
                    uuidMap.remove(uuid).execute(msg);
                } else {
                    logger.info("can't find matched uuid {}", uuid);
                }
            }
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (logger.isInfoEnabled()) {
            logger.info("channelInactive: [{}]", ctx.channel().remoteAddress());
        }
        unregisterCtxMBean(ctx);
        if (null != eventSessionClosed) {
            eventBus.fireEvent(eventSessionClosed, ctx.channel());
        }
        Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
        if (null != endpoint) {
            endpoint.stop();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("channelActive:{}", ctx.channel());
        if (null != eventSessionOpened) {
            eventBus.fireEvent(eventSessionOpened, ctx.channel());
        }
        try {
            //	create endpoint
            Endpoint endpoint = endpointFactory.createEndpoint(ctx.channel());
            if (null != endpoint) {
                TransportUtils.attachEndpointToSession(ctx, endpoint);
            }

            //	register mbean for ctx
            registerCtxMBean(ctx);
        } catch (Exception ex) {
            logger.warn(" createEndpoint:", ex);
            ctx.channel().close();
        }
    }

    private void registerCtxMBean(final ChannelHandlerContext ctx) {
        final Channel channel = ctx.channel();

        this._mbeanSupport.registerMBean(genChannelSuffix(channel), new ChannelMXBean() {

            @Override
            public String getId() {
                return channel.toString();
            }

            @Override
            public boolean isConnected() {
                return channel.isActive();
            }

            @Override
            public String getRemoteIp() {
                return ((InetSocketAddress) channel.remoteAddress()).getAddress().getHostAddress();
            }

            @Override
            public int getRemotePort() {
                return ((InetSocketAddress) channel.remoteAddress()).getPort();
            }

            @Override
            public String getLocalIp() {
                return ((InetSocketAddress) channel.localAddress()).getAddress().getHostAddress();
            }

            @Override
            public int getLocalPort() {
                return ((InetSocketAddress) channel.localAddress()).getPort();
            }

			@Override
			public String invokeRemoteMBeanOperation(
					final String objectName,
					final String operationName, 
					final String[] params) {
                final Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
                final Object waiter = new Object();
                final AtomicReference<JSONObject> retref = new AtomicReference<JSONObject>(null);
                if ( endpoint instanceof AbstractEndpoint ) {
                	((AbstractEndpoint)endpoint).sendActionWithFeedback("invokeMBean", new JSONObject() {{
                		this.put("objectName", objectName);
                		this.put("operationName", operationName);
                		this.put("params", params);
                	}}, new Visitor<JSONObject>() {

						@Override
						public void visit(final JSONObject t) throws Exception {
							retref.set(t);
		                	synchronized(waiter) {
		                		waiter.notify();
		                	}
						}});
                	
                	synchronized(waiter) {
                		try {
							waiter.wait( 1000L * 10);
							return	JSON.toJSONString( retref.get() );
							
						} catch (InterruptedException e) {
							logger.warn("exception when wait for invoke remote mbean operation, {}.{}, detail: {}", 
									objectName, operationName, ExceptionUtils.exception2detail(e));
							return	"exception when wait for invoke remote mbean operation, " + objectName + "." + operationName;
						}
                	}
                }
				return null;
			}
        });

    }

    private void unregisterCtxMBean(final ChannelHandlerContext ctx) {
        final Channel channel = ctx.channel();

        this._mbeanSupport.unregisterMBean(genChannelSuffix(channel));
    }

    /**
     * @param channel
     * @return
     */
    private static String genChannelSuffix(final Channel channel) {
        InetSocketAddress dstAddr = (InetSocketAddress) channel.remoteAddress();
        return "id=" + dstAddr.getAddress().getHostAddress() + "_" + dstAddr.getPort();
    }

    public String getEventRequestReceived() {
        return eventRequestReceived;
    }

    public void setEventRequestReceived(String eventRequestReceived) {
        this.eventRequestReceived = eventRequestReceived;
    }

    public String getEventSessionOpened() {
        return eventSessionOpened;
    }

    public void setEventSessionOpened(String eventSessionOpened) {
        this.eventSessionOpened = eventSessionOpened;
    }

    public String getEventSessionClosed() {
        return eventSessionClosed;
    }

    public void setEventSessionClosed(String eventSessionClosed) {
        this.eventSessionClosed = eventSessionClosed;
    }

    public void setEventBus(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    public Map<Object, Closure> getUuidMap() {
        return uuidMap;
    }

    /**
     * 必须为线程安全的,内存安全的Map,建议使用guava的Maps类创建
     *
     * @param uuidMap
     */
    public void setUuidMap(Map<Object, Closure> uuidMap) {
        this.uuidMap = uuidMap;
    }

    public void setEndpointFactory(EndpointFactory endpointFactory) {
        this.endpointFactory = endpointFactory;
    }
}
